package com.feishi.core.util.queue.disruptor;


import com.feishi.core.util.queue.Producer;
import com.feishi.core.util.queue.Customer;
import com.lmax.disruptor.dsl.Disruptor;

/**
 * Created by caixq on 4/25 025.
 */
public class DisrupterUtil {
    public static <T> Producer<T> getProductor(Customer<T> customer, int bufferSize){

        Disruptor<DisruptorData<T>> disruptor = new Disruptor<DisruptorData<T>>(()->{return new DisruptorData<T>();}, bufferSize, r->{return new Thread(r);});
        disruptor.handleEventsWith(new CustomerImpl<T>(customer) );
        disruptor.start();
        ProducerImpl productor=new ProducerImpl(disruptor.getRingBuffer());
        return productor;
    }

    public static void main(String[] args) throws InterruptedException {
        //(str,sequence,endOfBatch)->System.out.println(str+"::::::"+sequence+"::::"+endOfBatch)
        Producer<String> productor=getProductor(new Customer<String>() {
            @Override
            public void execute(String str, long sequence, boolean endOfBatch) throws Exception {
                str+="::::::"+sequence+"::::"+endOfBatch;
                System.out.println(str);
                Thread.sleep(1000);
            }
        }, 128);

        for (long l = 0; true; l++) {
            String data="--"+l*1.5+"---";
            productor.push(data);
            System.out.println("put data::"+data);
            Thread.sleep(100);
        }
    }
}
